/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.flink.connector.jdbc;

import org.apache.flink.annotation.PublicEvolving;
import org.apache.flink.connector.jdbc.internal.GenericJdbcSinkFunction;
import org.apache.flink.connector.jdbc.internal.JdbcOutputFormat;
import org.apache.flink.connector.jdbc.internal.connection.SimpleJdbcConnectionProvider;
import org.apache.flink.connector.jdbc.internal.executor.JdbcBatchStatementExecutor;
import org.apache.flink.connector.jdbc.xa.JdbcXaSinkFunction;
import org.apache.flink.connector.jdbc.xa.XaFacade;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.util.function.SerializableSupplier;

import javax.sql.XADataSource;

import java.util.function.Function;

/** Facade to create JDBC {@link SinkFunction sinks}. */
@PublicEvolving
public class JdbcSink {

    /**
     * Create a JDBC sink with the default {@link JdbcExecutionOptions}.
     * JdbcSink 类的静态方法 sink的实现。
     * 该方法用于创建一个 SinkFunction对象，用于将数据写入 JDBC数据源。
     * @see #sink(String, JdbcStatementBuilder,JdbcExecutionOptions,JdbcConnectionOptions)
     */
    public static <T> SinkFunction<T> sink(
            String sql,
            JdbcStatementBuilder<T> statementBuilder,
            JdbcConnectionOptions connectionOptions) {
        
        return sink(sql, statementBuilder,JdbcExecutionOptions.defaults(),connectionOptions);
    }

    /**
     * Create a JDBC sink.
     * 
     * <p>Note: the objects passed to the return sink can be processed in batch and retried.
     * Therefore, objects can not be {@link
     * org.apache.flink.api.common.ExecutionConfig#enableObjectReuse() reused}.
     *
     * @param sql arbitrary DML query (e.g. insert, update, upsert)
     * @param statementBuilder sets parameters on {@link java.sql.PreparedStatement} according to
     *     the query
     * @param <T> type of data in {@link
     *     org.apache.flink.streaming.runtime.streamrecord.StreamRecord StreamRecord}.
     * @param executionOptions parameters of execution, such as batch size and maximum retries
     * @param connectionOptions parameters of connection, such as JDBC URL
     */
    // 首先接受 SQL语句、JdbcStatementBuilder对象、JdbcExecutionOptions对象
    // 和 JdbcConnectionOptions对象作为参数。
    public static <T> SinkFunction<T> sink(
            String sql,
            JdbcStatementBuilder<T> statementBuilder,
            JdbcExecutionOptions executionOptions,
            JdbcConnectionOptions connectionOptions) {
        
        // 将JdbcOutputFormat对象传递给GenericJdbcSinkFunction构造函数，
        // 并返回一个 SinkFunction 对象。
        return new GenericJdbcSinkFunction<>(
                // 然后，创建一个 JdbcOutputFormat对象，并使用传入的参数设置其相关参数。
                new JdbcOutputFormat<>(
                        // 提供 JDBC连接的对象。
                        new SimpleJdbcConnectionProvider(connectionOptions), //Jdbc连接包装类

                        // 指定执行选项的对象。
                        // 执行参数类，如重试次数、最大等待时间、最大等待条数
                        executionOptions,

                        //用于执行批量 SQL语句。
                        // 返回的就是这个 SimpleBatchStatementExecutor这里封装了最终写入逻辑
                        context ->
                        JdbcBatchStatementExecutor.simple(sql, statementBuilder, Function.identity()),

                        //将输入记录转换为 JdbcIn对象的记录提取器。
                        JdbcOutputFormat.RecordExtractor.identity()
                )
        );
    }

    /**
     * Create JDBC sink which provides exactly-once guarantee.
     *
     * <p>Note: the objects passed to the return sink can be processed in batch and retried.
     * Therefore, objects can not be {@link
     * org.apache.flink.api.common.ExecutionConfig#enableObjectReuse() reused}.
     *
     * @param sql arbitrary DML query (e.g. insert, update, upsert)
     * @param statementBuilder sets parameters on {@link java.sql.PreparedStatement} according to
     *     the query
     * @param <T> type of data in {@link
     *     org.apache.flink.streaming.runtime.streamrecord.StreamRecord StreamRecord}.
     * @param executionOptions parameters of execution, such as batch size and maximum retries
     * @param exactlyOnceOptions exactly-once options. Note: maxRetries setting must be strictly set
     *     to 0 for the created sink to work properly and not to produce duplicates. See issue
     *     FLINK-22311 for details.
     * @param dataSourceSupplier supplies the {@link XADataSource}
     */
    
    // 这段代码是 JdbcSink类的静态方法 exactlyOnceSink的实现。
    // 该方法用于创建一个 SinkFunction对象，用于将数据以恰好一次的语义写入 JDBC数据源。
    //
    //首先接受 SQL语句、JdbcStatementBuilder对象、JdbcExecutionOptions对象、
    // JdbcExactlyOnceOptions对象和 SerializableSupplier<XADataSource> 对象作为参数。

    //该方法用于创建一个 SinkFunction对象，并将其用于将数据以恰好一次的语义写入 JDBC数据源。
    // 它可以在 Flink程序中使用，用于对 JDBC数据源进行事务性写入。
    public static <T> SinkFunction<T> exactlyOnceSink(
            String sql,
            JdbcStatementBuilder<T> statementBuilder,
            JdbcExecutionOptions executionOptions,
            JdbcExactlyOnceOptions exactlyOnceOptions,
            SerializableSupplier<XADataSource> dataSourceSupplier) {

        // 然后，使用传入的参数创建一个 JdbcXaSinkFunction对象，并返回该对象。
        return new JdbcXaSinkFunction<>(
                sql,
                statementBuilder,
                XaFacade.fromXaDataSourceSupplier(
                        dataSourceSupplier,
                        exactlyOnceOptions.getTimeoutSec(),
                        exactlyOnceOptions.isTransactionPerConnection()),
                executionOptions,
                exactlyOnceOptions);
    }

    private JdbcSink() {}
}
